package priv.lg.struct

import org.apache.spark.sql.{Dataset, SparkSession}
import priv.lg.bean.BusInfo
import priv.lg.foreachWriter.MysqlWriter

object ProcessTimeServer {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName(ProcessTimeServer.getClass.getCanonicalName)
      .master("local[*]").getOrCreate()
    val sc = spark.sparkContext
    sc.setLogLevel("WARN")
    import spark.implicits._
    // 读取kafka报文
    val kafkaDf = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "hadoop1:9092")
      .option("subscribe", "lagou_bus_info")
      .load()
    val kafkaValueDf = kafkaDf.selectExpr("cast(value as string)")
    val kafkaDs: Dataset[BusInfo] = kafkaValueDf.as[String].map(msg => BusInfo(msg))
    kafkaDs.writeStream
      .foreach(new MysqlWriter)
      .outputMode("append")
      .start()
      .awaitTermination()
  }

}
